package com.xncoding.pos.service.thread;

import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONUtil;
import com.xncoding.pos.common.expection.GlobalException;
import com.xncoding.pos.dao.TbUserDao;
import com.xncoding.pos.model.TbUser;
import com.xncoding.pos.service.thread.original.MyThreadFromCallable;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * java8里线程的应用-异步任务编排
 * 在 Java 8 中, 新增加了一个包含 50 个方法左右的类: CompletableFuture， 提供了非常强大的Future 的扩展功能， 可以帮助我们简化异步编程的复杂性， 提供了函数式编程的能力， 可以通过回调的方式处理计算结果
 *
 * @Description CompletableService
 * @Author guoyb
 * @Date 2021/12/13 14:51
 */
@Slf4j
@Service
public class CompletableService {

    // 带run的方法，无入参，无返回值。
    // 带accept的方法，有入参，无返回值。
    // 带supply的方法，无入参，有返回值。
    // 带apply的方法，有入参，有返回值。
    // 带handle的方法，有入参，有返回值，并且带异常处理。
    // 以Async结尾的方法，都是异步的，否则是同步的。
    // 以Either结尾的方法，只需完成任意一个。
    // 以Both/Combine结尾的方法，必须所有都完成。


    // // 对应 U apply(T t) ，接收参数 T并支持返回值 U
    // public <U> CompletionStage<U> thenApply(Function<? super T,? extends U> fn);
    // public <U> CompletionStage<U> thenApplyAsync(Function<? super T,? extends U> fn);

    // // 不接收参数也不支持返回值
    // public CompletionStage<Void> thenRun(Runnable action);
    // public CompletionStage<Void> thenRunAsync(Runnable action);

    // // 接收参数但不支持返回值
    // public CompletionStage<Void> thenAccept(Consumer<? super T> action);
    // public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);

    //接收参数:  功能：当前任务正常完成以后执行，当前任务的执行结果可以作为下一任务的输入参数，无返回值.


    // thenAccept()处理正常结果；是有入参，但是没有返回值的
    // exceptional()处理异常结果；
    // thenApplyAsync()用于串行化另一个CompletableFuture；
    // anyOf()和allOf()用于并行化多个CompletableFuture。



    // supplyAsync执行CompletableFuture任务，支持返回值。
    // //使用内置线程ForkJoinPool.commonPool()，根据supplier构建执行任务
    // public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
    // //指定自定义线程，根据supplier构建执行任务
    // public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

    //      没有返回值。
    // //使用内置线程ForkJoinPool.commonPool()，根据runnable构建执行任务
    // public static CompletableFuture<Void> runAsync(Runnable runnable)
    // //指定自定义线程，根据runnable构建执行任务
    // public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)

    // 常量值作为CompletableFuture返回
    // //有时候是需要构建一个常量的CompletableFuture
    // public static <U> CompletableFuture<U> completedFuture(U value)


    //     public CompletableFuture<Void> thenRun(Runnable action)
    //     public CompletableFuture<Void> thenRunAsync(Runnable action)
    //     //action用指定线程池执行
    //     public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)


    // 任务完成则运行，依赖上一个任务的结果，有返回值
    // public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
    // public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
    // //fn用指定线程池执行
    // public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)


    @Resource
    CompletableFutureService completableFutureService;

    @Resource
    TbUserDao tbUserDao;

    //可以shutdown，awaitTermination方便处理一些所有线程全部执行完才能够调用的服务接口等。
    @Resource
    ExecutorService executorService;
    // 将两个异步计算合并为一个，这两个异步计算之间相互独立，同时第二个又依赖于第一个的结果
    // 等待Future集合种的所有任务都完成。
    // 仅等待Future集合种最快结束的任务完成（有可能因为他们试图通过不同的方式计算同一个值），并返回它的结果。
    // 通过编程方式完成一个Future任务的执行（即以手工设定异步操作结果的方式）。
    // 应对Future的完成时间（即当Future的完成时间完成时会收到通知，并能使用Future的计算结果进行下一步的的操作，不只是简单地阻塞等待操作的结果）

    @SneakyThrows
    public void completableFutureTest_anyDone() {
        CompletableFuture<String> createOrder = completableFutureService.doSomething1("create order");
        CompletableFuture<String> reduceAccount = completableFutureService.doSomething2("reduce account");
        CompletableFuture<String> saveLog = completableFutureService.doSomething3("save log");

        // 等待所其中任意一个任务执行完
        CompletableFuture<Object> objectCompletableFuture = CompletableFuture.anyOf(createOrder, reduceAccount, saveLog);
        // 获取任务的返回结果
        Object obj = objectCompletableFuture.get();
        log.info("completableFutureTest_anyDone-----打印日志------{}", obj);
    }


    @SneakyThrows
    public void completableFutureTest_allDone() {
        CompletableFuture<String> createOrder = completableFutureService.doSomething1("create order");
        CompletableFuture<String> reduceAccount = completableFutureService.doSomething2("reduce account");
        CompletableFuture<String> saveLog = completableFutureService.doSomething3("save log");

        // 等待所有任务都执行完
        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(createOrder, reduceAccount, saveLog);

        // 阻塞等待所有任务执行完成
        voidCompletableFuture.join();
        Void unused = voidCompletableFuture.get();

        // 获取每个任务的返回结果
        // String result = createOrder.get() + reduceAccount.get() + saveLog.get();

        log.info("result-----打印日志------{}", saveLog.get());
    }

    /**
     * 等待所有任务完成, 常用
     */
    public void ThreadAllDown() {
        log.info("所有任务执行-----打印日志------{}", DateUtil.now());

        // 1. 创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        List<Integer> list = Arrays.asList(1, 2, 3);
        // 2. 提交任务，并调用join()阻塞等待所有任务执行完成
        CompletableFuture
                .allOf(
                        list.stream().map(key ->
                                CompletableFuture.runAsync(() -> {
                                    // 睡眠一秒，模仿处理过程
                                    try {
                                        Thread.sleep(1000L);
                                    } catch (InterruptedException e) {
                                    }
                                    System.out.println("结果" + key);
                                }, executorService))
                                .toArray(CompletableFuture[]::new))
                .join();
        executorService.shutdown();

        log.info("所有任务已完成-----打印日志------{}", DateUtil.now());
    }


    // CompletableFuture更强大的功能是，多个CompletableFuture可以串行执行，
    // 例如，定义两个CompletableFuture，第一个CompletableFuture根据证券名称查询证券代码，
    // 第二个CompletableFuture根据证券代码查询证券价格，这两个CompletableFuture实现串行操作如下：

    /**
     * 任务完成则运行action，依赖上一个任务的结果，无返回值
     */
    @SneakyThrows
    public void completableFutureTest1() {
        // 第一个任务:
        CompletableFuture<Long> cfQuery = CompletableFuture.supplyAsync(() -> {
            Long id = getUserId();
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            log.info("任务1-----打印日志------{}", id);
            return id;
        });

        //普通的异步
        CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(5000);
                log.info("-----打印日志------runAsync");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Long aLong = cfQuery.get();
        System.out.println("supplyAsync = " + aLong);

        // 任务完成则运行action，依赖上一个任务的结果，无返回值
        cfQuery.thenAcceptAsync(id -> {
            TbUser tbUser = tbUserDao.queryById(id);
            log.info("thenAcceptAsync-----打印日志------{}", JSONUtil.toJsonStr(tbUser));
        });

    }

    private Long getUserId() {
        List<TbUser> all = tbUserDao.getAll();
        TbUser tbUser = all.get(0);
        return tbUser.getId();
    }

    /**
     * 异常的情况
     * @return
     */
    public CompletableFuture<String> asyncProcessWithHandling() {
        return CompletableFuture.supplyAsync(() -> {
            // 模拟业务逻辑
            if (Math.random() > 0.5) {
                throw new GlobalException("处理失败");
            }
            return "处理成功";
        }, executorService).exceptionally(throwable -> {
            // 异常处理：提供默认值或转换异常
            log.error("异步处理失败", throwable);

            if (throwable instanceof GlobalException) {
                return "业务异常默认值";
            }
            return "系统异常默认值";
        });
    }

    // CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    //             // 业务逻辑
    //             return fetchDataFromRemote();
    //         })
    //         .handle((result, throwable) -> {
    //             if (throwable != null) {
    //                 // 异常处理
    //                 log.error("数据获取失败", throwable);
    //                 return getCachedData(); // 降级策略
    //             }
    //
    //             // 正常结果处理
    //             return processResult(result);
    //         })
    //         .thenApply(processedResult -> {
    //             // 后续处理，这里不会收到异常
    //             return transformResult(processedResult);
    //         })
    //         .whenComplete((finalResult, throwable) -> {
    //             // 最终清理，无论成功失败都会执行
    //             if (throwable != null) {
    //                 log.error("任务链执行失败", throwable);
    //             }
    //             cleanupResources();
    //         });


    /**
     * 任务完成则运行，依赖上一个任务的结果，有返回值
     *
     * @return
     */
    @SneakyThrows
    public TbUser completableFutureTest2() {

        CompletableFuture<TbUser> future = CompletableFuture
                .supplyAsync(() -> {
                    MyThreadFromCallable myThreadFromCallable = new MyThreadFromCallable(tbUserDao,2L);
                    String  msg = null;
                    try {
                        msg = myThreadFromCallable.call();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    log.info("completableFutureTest2----打印日志------{}", msg);
                    return msg;
                })
                .thenApplyAsync(id -> tbUserDao.getAll().get(0));

        boolean done = future.isDone();
        log.info("done-----打印日志------{}", done);
        return future.get();
    }


    /**
     * 两个CompletableFuture并行执行完，然后执行action，不依赖上两个任务的结果，无返回值
     *
     * @return
     */
    @SneakyThrows
    public void completableFutureTest3() {

        //第一个异步任务，常量任务
        CompletableFuture<String> first = CompletableFuture.completedFuture("hello world");
        CompletableFuture<Void> future = CompletableFuture
                //第二个异步任务
                .supplyAsync(() -> "hello siting")
                // () -> System.out.println("OK") 是第三个任务
                .runAfterBothAsync(first, () -> System.out.println("OK"));

    }

    /**
     * 两个CompletableFuture并行执行完，然后执行action，依赖上两个任务的结果，无返回值
     */
    public void completableFutureTest4() {

        //第一个异步任务，常量任务
        CompletableFuture<String> first = CompletableFuture.completedFuture("hello world");
        CompletableFuture<String> future = CompletableFuture
                //第二个异步任务
                .supplyAsync(() -> "hello siting")
                // (w, s) -> System.out.println(s) 是第三个任务
                .thenCombineAsync(first, (s, w) -> {
                    System.out.println(s);
                    return "OK";
                });
        System.out.println(future.join());

    }


    /**
     * 两个异步任务都完成后进行下一阶段的操作，此时我们可以使用 thenCombine 合并两个没有依赖关系的任务并行执行，当两个任务都完成之后对其结果进行操作。
     */
    @SneakyThrows
    public void completableFutureTest9() {

        CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("我女朋友小樱正在做红烧排骨");
            return "红烧排骨";
        }).thenCombine(CompletableFuture.supplyAsync(() -> {
            System.out.println("我女朋友小葵正在做清蒸鲈鱼");
            return "清蒸鲈鱼";
        }), (t1, t2) -> { //两个任务都执行完之后才会执行
            System.out.println(t1 + "," + t2 + " 并行做好了，通知我开吃！！");
            return "吃完了----";
        });

        String s = stringCompletableFuture.get();
        log.info("thenCombine-----打印日志------{}", s);
    }


    /**
     * 两个CompletableFuture并行执行完，然后执行fn，依赖上两个任务的结果，有返回值
     */
    public void completableFutureTest5() {

        //第一个异步任务，常量任务
        CompletableFuture<Long> first = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return getUserId();
        });


        CompletableFuture<Void> future = CompletableFuture
                //第二个异步任务
                .supplyAsync(() -> "hello siting")
                // (w, s) -> System.out.println(s) 是第三个任务
                .thenAcceptBothAsync(first, (res, id) -> {
                    TbUser tbUser = tbUserDao.queryById(id);
                    log.info("任务二结果-----打印日志------{}", res);
                    log.info("任务一结果-----打印日志------{}", id);
                    log.info("thenAcceptBothAsync-----打印日志------{}", JSONUtil.toJsonStr(tbUser));

                });

        //任务会阻塞
        Void join = future.join();

    }

    /**
     * 线程并行执行，谁先执行完则谁触发下一任务（二者选其最快）
     */
    public void completableFutureTest6() {
        //第一个异步任务，休眠2秒，保证最晚执行晚
        CompletableFuture<String> first = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (Exception e) {
            }
            log.info("保证最晚执行晚-------hello world");
            return "hello world";
        });
        CompletableFuture<Void> future = CompletableFuture
                //第二个异步任务不休眠
                .supplyAsync(() -> {
                    log.info("最快执行-----hello siting");
                    return "hello siting";
                })
                //() ->  System.out.println("OK") 是第三个任务
                .runAfterEitherAsync(first, () -> log.info("completableFutureTest6------OK"));


    }


    /**
     * 上一个任务或者other任务完成, 运行action，依赖最先完成任务的结果，无返回值
     */
    public void completableFutureTest7() {
        //第一个异步任务，休眠2秒，保证最晚执行晚
        CompletableFuture<String> first = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (Exception e) {
            }
            return "hello world";
        });
        CompletableFuture<Void> future = CompletableFuture
                //第二个异步任务
                .supplyAsync(() -> {
                    try {
                        TimeUnit.SECONDS.sleep(2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return DateUtil.now();
                })
                .acceptEitherAsync(first, data -> {
                    log.info("依赖最先完成任务的结果-----打印日志------{}----", data);
                    //    执行业务逻辑
                });

        // future.join();
    }

    /**
     * 上一个任务或者other任务完成, 运行fn，依赖最先完成任务的结果，有返回值
     * 哪个先返回就用那个的返回值.
     */
    @SneakyThrows
    public void completableFutureTest8() {

        //第一个异步任务，休眠1秒，保证最晚执行晚
        CompletableFuture<Long> first = CompletableFuture.supplyAsync(() -> {
            Long userId = null;
            try {
                Thread.sleep(3000);
                userId = getUserId();
            } catch (Exception e) {
            }
            return userId;
        });
        CompletableFuture<TbUser> future = CompletableFuture
                //第二个异步任务
                .supplyAsync(() -> 9L)
                // data ->  System.out.println(data) 是第三个任务
                .applyToEitherAsync(first, id -> {

                    log.info("依赖最先完成任务的结果，有返回值-----打印日志------{}", id);
                    TbUser tbUser = tbUserDao.queryById(id);
                    return tbUser;
                });

        TbUser tbUser = future.get();
        log.info("返回值-----打印日志------{}", JSONUtil.toJsonStr(tbUser));


        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            // 模拟耗时操作
            return "Result";
        });

        try {
            String result = completableFuture.get(); // 使用get()方法
            System.out.println(result);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println("Interrupted");
        } catch (ExecutionException e) {
            Throwable cause = e.getCause();
            System.out.println("Exception occurred: " + cause.getMessage());
        }

        try {
            String result = completableFuture.join(); // 使用join()方法
            System.out.println(result);
        } catch (CompletionException e) {
            Throwable cause = e.getCause();
            System.out.println("Exception occurred: " + cause.getMessage());
        }


    }

    /**
     * 两种渠道完成同一个事情，就可以调用这个方法，找一个最快的结果进行处理，最终有返回值。
     */
    public void completableFutureTest10() {
        String result = CompletableFuture.supplyAsync(new Supplier<String>() {
            @SneakyThrows
            @Override
            public String get() {
                Thread.sleep(1);
                return "Hi Boy";
            }
        }).applyToEither(CompletableFuture.supplyAsync(new Supplier<String>() {
            @SneakyThrows
            @Override
            public String get() {
                //   Thread.sleep(300);
                return "Hi Girl";
            }
        }), new Function<String, String>() {
            @Override
            public String apply(String s) {
                return s;
            }
        }).join();
        System.out.println(result);
    }


    // 假设有三个操作step1、step2、step3存在依赖关系，其中step3的执行依赖step1和step2的结果。
    public void completableFutureTest11() {
        //创建线程池，防止主线程结束造成其他线程终止运行，可以指定线程池中线程的数量
        ExecutorService executor = Executors.newFixedThreadPool(2);
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行step 1");
            return "step1 result";
        }, executor);
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行step 2");
            return "step2 result";
        });
        cf1.thenCombine(cf2, (result1, result2) -> {
            System.out.println(result1 + " , " + result2);
            System.out.println("执行step 3");
            return "step3 result";
        }).thenAccept(result3 -> System.out.println(result3));

        //关闭线程池
        executor.shutdown();
    }
}
